package net.sina.realtime.traffic.job;

import net.sina.realtime.traffic.bean.TrafficEvent;
import net.sina.realtime.traffic.function.TrafficMapFunction;
import net.sina.realtime.traffic.utils.DateTimeUtil;
import net.sina.realtime.traffic.utils.JdbcUtil;
import net.sina.realtime.traffic.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
public class JobTrafficYongDuJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<String> kafka = KafkaUtil.consumerKafka(env, "traffic-log");
        handle(kafka);

        env.execute("TransportationJob");
    }

    private static void handle(DataStream<String> kafka) {
        DataStream<TrafficEvent> watermarks = kafka
                .map(new TrafficMapFunction())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TrafficEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner( (event, ts) -> event.getTs()));

        // 全局窗口（不按 roadId 分组）
        DataStream<String> metrics = watermarks
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(new ProcessAllWindowFunction<TrafficEvent, String, TimeWindow>() {
                    final double FREE_FLOW_SPEED = 80.0;
                    int totalFlow = 0;
                    double totalSpeed = 0.0;
                    int congested = 0;
                    @Override
                    public void process( Context context, Iterable<TrafficEvent> events, Collector<String> out) {

                        String window_start_time = DateTimeUtil.convertLongToString(context.window().getStart(),DateTimeUtil.DATE_TIME_FORMAT);
                        String window_end_time = DateTimeUtil.convertLongToString(context.window().getEnd(),DateTimeUtil.DATE_TIME_FORMAT);

                        // 计算总车流量
                        Long timestamp = 0L;
                        for (TrafficEvent event : events) {
                            totalFlow++;
                            // 总车速
                            totalSpeed += event.getSpeed();
                            if (event.getSpeed() < FREE_FLOW_SPEED * 0.5) {
                                // 拥堵车辆数
                                congested++;
                            }
                            timestamp = event.getTs();
                        }

                        // 计算 平均车速 和 拥堵率
                        double avgSpeed = totalFlow == 0 ? 0 : totalSpeed / totalFlow;
                        double congestionRate = totalFlow == 0 ? 0 : (double) congested / totalFlow * 100;
                        String level = congestionRate > 60 ? "严重拥堵" :
                                congestionRate > 30 ? "中度拥堵" : "畅通";

                        // 输出汇总结果
                        String ss =
                                window_start_time + "," +
                                window_end_time + "," +
                                totalFlow  + "," +
                                Math.round(avgSpeed * 100) / 100.0 + "," +
                                Math.round(congestionRate * 100) / 100.0 + "," +
                                level + "," +
                                timestamp;
                        out.collect(ss);
                        System.out.println(ss);
                    }
                });

        // 数据写入到 ClickHouse
        JdbcUtil.sinkToClickhouseUpsert(metrics,
                "insert into traffic_log_report.dws_city_overview(\n" +
                        "    window_start_time, " +
                        "    window_end_time, " +
                        "    totalFlow, " +
                        "    avgSpeed, " +
                        "    congestionRate," +
                        "    congestionLevel," +
                        "    ts)\n" +
                        "VALUES (?,?,?,?,?,?,?)");
    }
}
